package rpc

import (
	"bufio"
	"context"
	"io"
	"log"
	"net"
	"time"
)

func NewStdServer(addr string, router HttpRouter) *StdServer {
	return &StdServer{
		addr:         addr,
		connected:    0,
		disconnected: 0,
		router:       router,
	}
}

type StdServer struct {
	addr         string
	connected    int32
	disconnected int32
	router       HttpRouter
}

func (s *StdServer) Start() {
	log.Printf("[http]starting on %s\n", s.addr)
	tcpAddr, err := net.ResolveTCPAddr("tcp", s.addr)
	if err != nil {
		log.Fatalf("resolve tcp error: %s", err)
	}
	ctx, _ := context.WithCancel(context.Background())
	// Try to bind
	list, err := net.Listen("tcp", tcpAddr.String())
	if err != nil {
		panic(err)
	}

	// Create stream
	stream := &TCPStreamLayer{
		advertise: tcpAddr,
		listener:  list.(*net.TCPListener),
	}

	// Verify that we have a usable address
	addr, ok := stream.Addr().(*net.TCPAddr)
	if !ok {
		_ = list.Close()
	}
	if addr.IP == nil {
		_ = list.Close()
	}

	log.Printf("[http]start on %s success\n", tcpAddr.String())
	s.listen(ctx, stream)
}

func (s *StdServer) listen(ctx context.Context, stream *TCPStreamLayer) {
	const baseDelay = 5 * time.Millisecond
	const maxDelay = 1 * time.Second

	var loopDelay time.Duration
	for {
		// Accept incoming connections
		conn, err := stream.Accept()
		if err != nil {
			if loopDelay == 0 {
				loopDelay = baseDelay
			} else {
				loopDelay *= 2
			}
			if loopDelay > maxDelay {
				loopDelay = maxDelay
			}
			select {
			case <-time.After(loopDelay):
				continue
			}
		}
		// No error, reset loop delay
		loopDelay = 0
		code := HttpCodecPool.Get().(*HttpCodec)
		code.Reset()
		// Handle the connection in dedicated routine
		go s.handleConn(ctx, conn, code)
	}
}

var HttpBufferSize = 64 * 1024 // 64KB

func (s *StdServer) handleConn(connCtx context.Context, conn net.Conn, code *HttpCodec) {
	r := bufio.NewReaderSize(conn, HttpBufferSize)
	for {
		if err := s.handleCommand(r, conn, code); err != nil {
			if err != io.EOF {
				log.Println("failed to decode incoming command", "error", err)
			}
			break
		}
	}
	_ = conn.Close()
	HttpCodecPool.Put(code)
}

func (s *StdServer) handleCommand(r *bufio.Reader, conn net.Conn, hc *HttpCodec) error {
	if hc.Finish {
		n, err := r.Read(hc.TmpBuf[0:])
		if err != nil {
			hc.Reset()
			return err
		}
		hc.Head.Write(hc.TmpBuf[:n])
		headerOffset, miss := hc.Parser.Parse(hc.Head.Bytes()[:n])
		for miss == ErrMissingData {
			m, read := r.Read(hc.TmpBuf[0:])
			if read != nil {
				hc.Reset()
				return err
			}
			n += m
			hc.Head.Write(hc.TmpBuf[:m])
			headerOffset, miss = hc.Parser.Parse(hc.Head.Bytes()[:n])
		}
		if miss != nil && miss != ErrMissingData {
			hc.Reset()
			return miss
		}
		hc.Body.Write(hc.Head.Bytes()[headerOffset:n])
	}
	err := fetchBody(r, hc)
	if err != nil {
		hc.Reset()
		return err
	}
	if hc.Finish {
		s.router.Publish(hc, hc.Body.Bytes())
		_, err1 := hc.OutBuf.WriteTo(conn)
		hc.Reset()
		if err1 != nil {
			return err1
		}
	}
	return nil
}

func fetchBody(r *bufio.Reader, hc *HttpCodec) error {
	for hc.Body.Len() < hc.Parser.ContentLength() {
		n, err := r.Read(hc.TmpBuf[0:])
		if err != nil {
			return err
		}
		if n == 0 {
			break
		}
		hc.Body.Write(hc.TmpBuf[:n])
	}
	if hc.Body.Len() < hc.Parser.ContentLength() {
		hc.Finish = false
	} else if hc.Body.Len() == hc.Parser.ContentLength() {
		hc.Finish = true
	} else {
		panic("message error")
	}
	return nil
}

// TCPStreamLayer implements StreamLayer interface for plain TCP.
type TCPStreamLayer struct {
	advertise net.Addr
	listener  *net.TCPListener
}

// Dial implements the StreamLayer interface.
func (t *TCPStreamLayer) Dial(address string, timeout time.Duration) (net.Conn, error) {
	return net.DialTimeout("tcp", address, timeout)
}

// Accept implements the net.Listener interface.
func (t *TCPStreamLayer) Accept() (c net.Conn, err error) {
	return t.listener.Accept()
}

// Close implements the net.Listener interface.
func (t *TCPStreamLayer) Close() (err error) {
	return t.listener.Close()
}

// Addr implements the net.Listener interface.
func (t *TCPStreamLayer) Addr() net.Addr {
	// Use an advertise addr if provided
	if t.advertise != nil {
		return t.advertise
	}
	return t.listener.Addr()
}
